package com.example.toy.core.utils;

import com.example.toy.core.commit.CommitMsg;
import com.example.toy.core.commit.CommitType;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema;
import org.apache.commons.lang.StringUtils;

/**
 * SchemaUtils
 *
 * @author shenbing
 * @date 2021/12/20
 */
@Slf4j
public class SchemaHelper {

  public static String toString(Schema schema) {
    return schema.toString();
  }

  /**
   *  convert string to schema
   * @param str
   * @return null if str is blank, else the schema
   */
  public static Schema fromString(String str) {
    if (StringUtils.isBlank(str)) {
      return null;
    }
    Schema.Parser parser = new Schema.Parser();
    return parser.parse(str);
  }

  public static Schema.Type findUnionElementType(Schema schema) {
    if (schema.getType() != Schema.Type.UNION) {
      throw new IllegalArgumentException(
          "Unsupported schema type cast as Union: " + schema.getType());
    }
    List<Schema> types = schema.getTypes();
    for (Schema type : types) {
      if (type.getType() != Schema.Type.NULL && type.getType() != Schema.Type.UNION) {
        return type.getType();
      }
    }
    throw new IllegalArgumentException(
        "Can not find available type for schema type: " + schema.getType());
  }

  public static Object convert(Object value, Schema schema) {
    boolean nullable = schema.isNullable();
    Schema.Type elementType = schema.isUnion() ? findUnionElementType(schema) : schema.getType();

    switch (elementType) {
      case STRING:
        return convertToString(value, nullable);
      case BYTES:
        return convertToBytes(value, nullable);
      case INT:
        return convertToInt(value, nullable);
      case LONG:
        return convertToLong(value, nullable);
      case FLOAT:
        return convertToFloat(value, nullable);
      case DOUBLE:
        return convertToDouble(value, nullable);
      case BOOLEAN:
        return convertToBoolean(value, nullable);
      case NULL:
      case RECORD:
      case ENUM:
      case ARRAY:
      case MAP:
      case UNION:
      case FIXED:
        throw new IllegalArgumentException("Unsupported convert value to schema: " + elementType);
      default:
        log.warn("Can not convert value to unknown schema type, return value as default.");
        return value;
    }
  }

  private static Object convertToBoolean(Object value, boolean nullable) {
    if (value == null) {
      return nullable ? null : false;
    }
    if (value.getClass() == Boolean.class || value.getClass() == Boolean.TYPE) {
      return value;
    } else {
      return Boolean.parseBoolean(String.valueOf(value));
    }
  }

  private static Object convertToDouble(Object value, boolean nullable) {
    if (value == null) {
      return nullable ? null : 0D;
    }
    if (value.getClass() == Double.class || value.getClass() == Double.TYPE) {
      return value;
    } else {
      return Double.parseDouble(String.valueOf(value));
    }
  }

  private static Object convertToFloat(Object value, boolean nullable) {
    if (value == null) {
      return nullable ? null : 0F;
    }
    if (value.getClass() == Float.class || value.getClass() == Float.TYPE) {
      return value;
    } else {
      return Float.parseFloat(String.valueOf(value));
    }
  }

  private static Object convertToLong(Object value, boolean nullable) {
    if (value == null) {
      return nullable ? null : 0L;
    }
    if (value.getClass() == Long.class || value.getClass() == Long.TYPE) {
      return value;
    } else {
      return Long.parseLong(String.valueOf(value));
    }
  }

  private static Object convertToBytes(Object value, boolean nullable) {
    if (value == null) {
      return nullable ? null : new byte[0];
    }
    if (value.getClass().isArray()) {
      Class<?> componentType = value.getClass().getComponentType();
      if (componentType == Byte.class) {
        return value;
      } else if (componentType == Byte.TYPE) {
        return value;
      }
    }
    throw new IllegalArgumentException("Unsupported convert value to byte array");
  }

  private static final String EMPTY = "";

  private static String convertToString(Object value, boolean nullable) {
    if (value == null) {
      return nullable ? null : EMPTY;
    }
    return String.valueOf(value);
  }

  private static Object convertToInt(Object value, boolean nullable) {
    if (value == null) {
      return nullable ? null : 0;
    }
    Class<?> clz = value.getClass();
    if (clz == Integer.class) {
      return value;
    } else if (clz == Integer.TYPE) {
      return value;
    } else {
      return Integer.parseInt(String.valueOf(value));
    }
  }

  public static Schema mergeSchema(Schema initSchema, List<CommitMsg> alterColumnCommits) {
    if (alterColumnCommits.isEmpty()) {
      return initSchema;
    }
    for (CommitMsg alterCommit : alterColumnCommits) {
      if (alterCommit.getCommitType() != CommitType.ALTER_COLUMN) {
        throw new IllegalArgumentException("Can not merge schema from commit: " + alterCommit);
      }
      // TODO
    }
    return initSchema;
  }

  //TODO
  public static Schema mergeSchema(Schema initSchema, Schema addColumns) {
    return initSchema;
  }
}
